package com.example.springtask.config;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.integration.async.AsyncItemProcessor;
import org.springframework.batch.integration.async.AsyncItemWriter;
import org.springframework.batch.integration.chunk.RemoteChunkingWorkerBuilder;
import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.kafka.dsl.Kafka;
import org.springframework.integration.kafka.dsl.KafkaMessageDrivenChannelAdapterSpec;
import org.springframework.integration.kafka.inbound.KafkaMessageSource;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConsumerProperties;
import org.springframework.scheduling.support.PeriodicTrigger;

import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

@Profile("worker")
@Configuration
@EnableBatchIntegration
@EnableBatchProcessing
public class WorkerConfiguration {

    @Autowired
    private RemoteChunkingWorkerBuilder<String,String> workerBuilder;

    @Bean
    public IntegrationFlow workerFlow() {
        return this.workerBuilder
                .itemProcessor(asynProcessor())
                .itemWriter(asynWriter())
                .inputChannel(replies()) // requests received from the master
                .outputChannel(requests()) // replies sent to the master
                .build();
    }

    @Bean
    public DirectChannel requests() {
        return new DirectChannel();
    }

    @Bean
    public QueueChannel replies() {
        return new QueueChannel();
    }

    @Bean
    public IntegrationFlow inboundFlow(ConsumerFactory consumerFactory){
        ConsumerProperties consumerProperties = new ConsumerProperties("requests");
        return IntegrationFlows.from(
                Kafka.inboundChannelAdapter(consumerFactory, consumerProperties),
                c->c.poller(Pollers.fixedRate(1000).maxMessagesPerPoll(1))
        ).channel(replies()).get();
//        return IntegrationFlows.from(
//               Kafka.messageDrivenChannelAdapter(consumerFactory, "requests")
//        ).channel(replies()).get();
    }


    @Bean
    public IntegrationFlow outboundFlow(KafkaTemplate kafkaTemplate){
        return IntegrationFlows.from(requests())
                .handle(Kafka.outboundChannelAdapter(kafkaTemplate).topic("replies"))
                .get();
    }

    @Bean
    public ItemProcessor asynProcessor() {
        ItemProcessor itemProcessor = s->{
            System.out.println("processor" +s.toString());
            return s;
        };
        AsyncItemProcessor asyncItemProcessor = new AsyncItemProcessor<>();
        asyncItemProcessor.setTaskExecutor(new SyncTaskExecutor());
        asyncItemProcessor.setDelegate(itemProcessor);
        return asyncItemProcessor;
    }


    @Bean
    public AsyncItemWriter asynWriter() {
        ItemWriter itemWriter = l->{
            System.out.println("writer" + l);
        };
        AsyncItemWriter asyncItemWriter = new AsyncItemWriter<>();
        asyncItemWriter.setDelegate(itemWriter);
        return asyncItemWriter;
    }

    @Bean(name = PollerMetadata.DEFAULT_POLLER)
    public PollerMetadata poller() {
        return Pollers.fixedRate(500).get();
    }
}
